import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Date;
import java.util.Properties;
import java.util.Random;

/**
 * 一个自定义分区的Kafka Producer类，传入两个参数：
 * topic num
 * 设置主题和message条数
 *
 * 模拟用户点击日志，日志格式为:“时间,网址,IP地址"格式
 *
 * 自定义分区，通过IP地址最后一位与分区数求余，message分散到0~partition - 1这些分区中
 *
 * 执行过程：
 * 1、创建一个topic
 * kafka-topic.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 3 --topic xxxx
 * 2、运行本类中的代码
 * 3、查看message
 * kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic xxxx
 * kafka
 */
public class SimpleKafkaProducer
{
    /**
     * Producer的两个泛型，第一个指定Key的类型，第二个指定value的类型
     */
    private static Producer<String, String> producer;

    public SimpleKafkaProducer()
    {
        Properties props = new Properties();
        /**
         * 指定producer连接的broker列表
         */
//        props.put("metadata.broker.list", "m000:9092, m001:9092, m002:9092");
        props.put("metadata.broker.list", "localhost:9092");
        /**
         *
         * 指定message的序列化方法，用户可以通过实现kafka.serializer.Encoder接口自定义该类
         * 默认情况下message的key和value都用相同的序列化，但是可以使用"key.serializer.class"指定key的序列化
         */
        props.put("serializer.class", "kafka.serializer.StringEncoder");
        /**
         * 这个参数用于通知broker接收到message后是否向producer发送确认信号
         *  0 - 表示producer不用等待任何确认信号，会一直发送消息
         *  1 - 表示leader状态的replica在接收到message后需要向producer发送一个确认信号，否则producer进入等待状态
         * -1 - 表示leader状态的replica需要等待所有in-sync状态的replica都接收到消息后才会向producer发送确认信号，再次之前producer一直处于等待状态
         */
        props.put("request.required.acks", "1");
        /**
         * 指定partition类，自定义的分区类，继承自kafka.producer.Partitioner接口
         */
//        props.put("partitioner.class", "ckm.kafka.producer.SimplePartitioner");
        ProducerConfig config = new ProducerConfig(props);

        producer = new Producer<String, String>(config);
    }

    public static void main(String[] args)
    {
//        if (args.length < 2) {
//            System.out.println("Please Input Topic and Message Numbers");
//        }
        String topic = "test";
        int count = 8;
        System.out.println("Topic = " + topic);
        System.out.println("Message Nums = " + count);

        SimpleKafkaProducer simpleProducer = new SimpleKafkaProducer();
        simpleProducer.publishMessage(topic, count);
    }

    /**
     * 根据topic和消息条数发送消息
     * @param topic
     * @param count
     */
    private void publishMessage(String topic, int count)
    {
        Random random = new Random();
        for (int i = 0; i < count; i ++)
        {
            String runtime = new Date().toString();
            // 访问的IP地址
            String clientIP = "192.168.1." + random.nextInt(255);
            String msg = runtime + ",kafka.apache.org," + clientIP;
            System.out.println("msg = " + msg);
            /**
             * 第一个泛型指定用于分区的key的类型，第二个泛型指message的类型
             * topic只能为String类型
             * 和上一个Producer相比，多了一个用于分区的key
             */
            KeyedMessage<String, String> data = new KeyedMessage<String, String>(topic, clientIP, msg);
            producer.send(data);
        }
        producer.close();
    }
}

//reference:
//https://blog.csdn.net/dabokele/article/details/52215968